package com.enba.boot.mq.sender.rocketmq;

import com.alibaba.fastjson.JSON;
import com.enba.boot.mq.MessageInteface;
import com.enba.boot.mq.MessageSenderStrategy;
import com.enba.boot.mq.properties.MqProperties;
import com.enba.boot.mq.sender.activemq.ActiveMqMessageBase;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;

/** Rocketmq 集群模式消息发送 */
public class RocketmqClusterMessageSenderStrategy
    implements MessageSenderStrategy<ActiveMqMessageBase> {

  @Autowired private RocketMQTemplate rocketMQTemplate;

  @Autowired private MqProperties mqProperties;

  @Override
  public String getStrategyKey() {
    return mqProperties.getRocketmq().getStrategyKey();
  }

  @Override
  public void sendMessageByQueue(String destinationName, String message) {
    logger.info(
        "RocketmqClusterMessageSenderStrategy.sendMessage destinationName:{}, message:{}",
        destinationName,
        message);
    rocketMQTemplate.convertAndSend(destinationName, message);
  }

  @Override
  public void sendMessageByTopic(String destinationName, String message) {
    logger.info(
        "RocketmqClusterMessageSenderStrategy.sendMessageByTopic destinationName:{}, message:{}",
        destinationName,
        message);
    rocketMQTemplate.convertAndSend(destinationName, message);
  }

  @Override
  public void sendMessageByTopic(String topicName, String msg, String key) {
    logger.info(
        "RocketmqClusterMessageSenderStrategy.sendMessageByTopic topicName:{}, message:{},key:{}",
        topicName,
        msg,
        key);

    // 创建消息对象，并设置消息key
    Message<String> message =
        MessageBuilder.withPayload(msg)
            .setHeader("KEYS", key) // 设置消息key
            .build();

    rocketMQTemplate.send(topicName, message);
  }

  @Override
  public void sendMessageByQueue(
      String destinationName, MessageInteface<ActiveMqMessageBase> messageInteface) {
    // 消息业务唯一标识
    String bizId = messageInteface.getBizId();
    // 消息业务内容
    ActiveMqMessageBase bizBody = messageInteface.getBizBody();

    logger.info(
        "RocketmqClusterMessageSenderStrategy.sendMessage bizId:{}, bizBody:{}", bizId, bizBody);

    rocketMQTemplate.convertAndSend(destinationName, messageInteface);
  }

  @Override
  public void sendMessageByTopic(
      String destinationName, MessageInteface<ActiveMqMessageBase> messageInteface) {
    logger.info(
        "RocketmqClusterMessageSenderStrategy.sendMessageByTopic destinationName:{}, message:{}",
        destinationName,
        JSON.toJSONString(messageInteface));
    rocketMQTemplate.convertAndSend(destinationName, messageInteface);
  }
}
